package com.core.image.core;
import com.core.image.core.box.*;
import com.core.image.core.impl.ScheduleJob;
import com.core.image.core.impl.Scheduler;
import com.core.image.core.impl.SocketChannelAdapter;
import com.core.image.core.impl.async.AsncReceiveDispatcher;
import com.core.image.core.impl.async.AsyncSendDispatcher;
import com.core.image.core.impl.bridge.BridgeSocketDispatcher;
import com.core.image.utils.CloseUtils;


import java.io.*;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

public abstract class Connector implements Closeable, SocketChannelAdapter.OnChannelStatusChangedListener {
    private UUID key = UUID.randomUUID();
    private SocketChannel channel;
    private Sender sender;
    private Receiver receiver;
    private SendDispatcher sendDispatcher;
    private ReceiveDispacher receiveDispacher;
    private final List<ScheduleJob> scheduleJobs = new ArrayList<>(4);
    public void setup(SocketChannel socketChannel) throws IOException {
        this.channel = socketChannel;
        IoContext context = IoContext.get();
        SocketChannelAdapter adapter = new SocketChannelAdapter(channel, context.getIoProvider(), this);
        this.sender = adapter;
        this.receiver = adapter;
        sendDispatcher = new AsyncSendDispatcher(sender);
        receiveDispacher = new AsncReceiveDispatcher(receiver,receivePacketCalback);
        receiveDispacher.start();
    }
    public void send(String msg){
            SendPacket packet = new StringSendPacket(msg);
            sendDispatcher.send(packet);
    }
    public void send(SendPacket packet){
        sendDispatcher.send(packet);
    }
    public void changeToBridge(){
       if (receiveDispacher instanceof BridgeSocketDispatcher){
           // 已改变直接返回
           return;
       }
       receiveDispacher.stop();
       BridgeSocketDispatcher dispatcher = new BridgeSocketDispatcher(receiver);
       receiveDispacher = dispatcher;
       receiveDispacher.start();
    }

    /**
     *  将另外一个链接的发送者绑定到当前链接的桥接调度器上实现两个链接的桥接功能
     * @param sender 另外一个链接的发送者
     */
    public void bindToBridge(Sender sender){
        if (sender == this.sender){
            throw new UnsupportedOperationException("Can not set current connector sender to self bridge mode!");
        }
        if (!(receiveDispacher instanceof BridgeSocketDispatcher)){
            throw new IllegalStateException("receiveDispatcher is not BridgeSocketDispatcher!");
        }
        ((BridgeSocketDispatcher) receiveDispacher).bindSender(sender);
    }

    /**
     * 将之前链接的发送者解除绑定，解除桥接数据发送功能
     */
    public void unBindToBridge(){
        if (!(receiveDispacher instanceof BridgeSocketDispatcher)) {
            throw new IllegalStateException("receiveDispatcher is not BridgeSocketDispatcher!");
        }

        ((BridgeSocketDispatcher) receiveDispacher).bindSender(null);
    }
    /**
     * 获取当前链接的发送者
     *
     * @return 发送者
     */
    public Sender getSender(){
        return sender;
    }
    /**
     * 发射一份空闲超时事件。
     */
    public void fireIdleTimeoutEvent(){
        sendDispatcher.sendHeartbeat();
    }
    @Override
    public void close() throws IOException {
        synchronized (scheduleJobs){
            for (ScheduleJob scheduleJob:scheduleJobs){
                scheduleJob.unSchedule();
            }
            scheduleJobs.clear();
        }
        receiveDispacher.close();
        sendDispatcher.close();
        sender.close();
        channel.close();
    }
    /**
     * 当接收包是文件时，需要得到一份空的文件用以数据存储
     *
     * @param length     长度
     * @param headerInfo 额外信息
     * @return 新的文件
     */
    protected abstract File createNewReceiveFile(long length, byte[] headerInfo);
    /**
     * 获取最后的活跃时间点
     *
     * @return 发送、接收的最后活跃时间
     */
    public long getLastActiveTime(){
        return Math.max(sender.getLastWriteTime(),receiver.LastReadTime());
    }
    /**
     * 发射一份异常事件，子类需要关注
     *
     * @param throwable 异常
     */
    public void fireExceptionCaught(Throwable throwable) {
    }
    public void schedule(ScheduleJob job){
        synchronized (scheduleJobs){
            if (scheduleJobs.contains(job)){
                return;
            }
        IoContext context = IoContext.get();
        Scheduler scheduler = context.getScheduler();
        job.schedule(scheduler);
        scheduleJobs.add(job);
        }

    }

    @Override
    public void onChannelClosed(SocketChannel channel) {
        CloseUtils.close(channel);

    }
    protected  void  onReceivePacket(ReceivePacket packet){
        System.out.println(getKey().toString()+":"+(String)packet.entity());
    }
    /**
     * 当接收包是直流数据包时，需要得到一个用以存储当前直流数据的输出流，
     * 所有接收到的数据都将通过输出流输出
     *
     * @param length     长度
     * @param headerInfo 额外信息
     * @return 输出流
     */
    public abstract OutputStream createNewReceiveDirectOutputStream(long length, byte[] headerInfo);

    protected abstract  File createNewFile();
    private ReceiveDispacher.ReceivePacketCalback receivePacketCalback = new ReceiveDispacher.ReceivePacketCalback() {
        @Override
        public void onReceivePacketCompleted(ReceivePacket packet) {
               onReceivePacket(packet);
        }

        @Override
        public ReceivePacket<?,?> onArriveNewPacket(byte type, long length, byte[] headerInfo) {
            switch (type){
                case Packet.TYPE_MEMORY_BYTES:
                    return new ByteReceivePacket(length);
                case Packet.TYPE_MEMORY_STRING:
                    return new StringRecevicePacket(length);
                case Packet.TYPE_STREAM_FILE:
                    return new FileReceivePacket(length,createNewReceiveFile(length,headerInfo));
                case Packet.TYPE_STREAM_DIRECT:
                     return new StreamDirectReceivePacket(createNewReceiveDirectOutputStream(length, headerInfo), length);
                default:
                    throw new UnsupportedOperationException("Unsupported packet type:" + type);
            }

        }

        @Override
        public void onReceivedHeartbeat() {
            System.out.println(key.toString() + ":[Heartbeat]");
        }


    };
    public UUID getKey() {
        return key;
    }


}
